package com.ruoyi.job.thread;

import com.alibaba.fastjson.JSONObject;
import com.api.equipment.RemoteEquipmentService;
import com.api.equipment.domain.LgTemperature;
import com.ruoyi.common.core.utils.SpringUtils;
import org.eclipse.paho.client.mqttv3.*;

import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: lek
 * @Description: TODO
 * @Date: 2025/3/13 19:48
 * @Version: 1.0
 */

public class TGThread extends Thread{
    private String serial;
    private String cmd;
    private Random random = new Random();
    String brokerUrl = "tcp://124.223.12.211:1883";
    String pubclientId = "temperature_publisher";
    String subclientId = "temperature_subscribe";
    String topic = "sensor/temgps/";
    int pubQos = 1;
    int subQos = 1;
    public static ConcurrentHashMap<String, Thread> consoleThreadMap = new ConcurrentHashMap<String, Thread>();
    public ConcurrentHashMap<String, JSONObject> subMessage = new ConcurrentHashMap<String, JSONObject>();

    private volatile boolean running = true; // 新增运行状态标志
    private MqttClient publisherClient;
    private MqttClient subscriberClient;

    private RemoteEquipmentService remoteEquipmentService = SpringUtils.getBean(RemoteEquipmentService.class);
    public String getSerial() {
        return serial;
    }

    public void setSerial(String serial) {
        this.serial = serial;
    }

    public static ConcurrentHashMap<String, Thread> getConsoleThreadMap() {
        return consoleThreadMap;
    }

    public static void setConsoleThreadMap(ConcurrentHashMap<String, Thread> consoleThreadMap) {
        TGThread.consoleThreadMap = consoleThreadMap;
    }

    public String getCmd() {
        return cmd;
    }

    public void setCmd(String cmd) {
        this.cmd = cmd;
    }

    public ConcurrentHashMap<String, JSONObject> getSubMessage() {
        return subMessage;
    }

    public void setSubMessage(ConcurrentHashMap<String, JSONObject> subMessage) {
        this.subMessage = subMessage;
    }
    @Override
    public void run() {
        try {
            if (cmd.equals("publish")) {
                consoleThreadMap.put("publish"+serial, TGThread.currentThread());
                System.out.println("publish线程开始: " + this);
                this.publish();
            } else if (cmd.equals("subscribe")) {
                consoleThreadMap.put("subscribe"+serial, TGThread.currentThread());
                System.out.println("sub线程开始: " + this);
                this.subscribe();
            }
        } catch (MqttException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            consoleThreadMap.remove(serial); // 确保清理
            System.out.println("线程终止: " + serial);
        }
    }

    // 发布方法改进
    public void publish() throws MqttException, InterruptedException {
        publisherClient = new MqttClient(brokerUrl, pubclientId + "_" + serial); // 唯一ClientID
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        publisherClient.connect(options);
        String targetTopic = topic + serial;

        try {
            while (running && !Thread.currentThread().isInterrupted()) {
                // 构造并发送消息
                JSONObject sensorData = new JSONObject();
                sensorData.put("temperature", 20.0 + random.nextInt(10));
                sensorData.put("humidity", 50 + random.nextInt(40));
                sensorData.put("latitude", 30.2741 + random.nextDouble() * 0.01);
                sensorData.put("longitude", 120.1551 + random.nextDouble() * 0.01);

                MqttMessage message = new MqttMessage(sensorData.toJSONString().getBytes());
                message.setQos(pubQos);
                publisherClient.publish(targetTopic, message);

                System.out.println("Published: " + sensorData);
                Thread.sleep(30000); // 允许被中断
            }
        } finally {
            if (publisherClient != null && publisherClient.isConnected()) {
                publisherClient.disconnectForcibly(1000); // 强制断开
            }
        }
    }

    // 订阅方法改进
    public void subscribe() throws MqttException, InterruptedException {
        subscriberClient = new MqttClient(brokerUrl, subclientId + "_" + serial);
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        String targetTopic = topic + serial;

        subscriberClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                System.out.println("Connection lost: " + cause.getMessage());
                running = false; // 触发终止
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                // 处理消息逻辑...
                String payload = new String(message.getPayload());
                JSONObject data = JSONObject.parseObject(payload); // Fastjson解析

                LgTemperature lgTemperature = remoteEquipmentService.selectLgVehicleBylicensePlate(serial);
                System.out.println(lgTemperature.toString());
                lgTemperature.setCurrentTemperature(data.getInteger("temperature"));
                lgTemperature.setCurrentHumidity(data.getInteger("humidity"));
                lgTemperature.setLatitudeX(data.getDouble("latitude"));
                lgTemperature.setLatitudeY(data.getDouble("longitude"));

                remoteEquipmentService.setTem(lgTemperature);

                System.out.println("\nReceived Sensor Data:");
                System.out.println("Temperature: " + data.getDouble("temperature") + "℃");
                System.out.println("Humidity: " + data.getInteger("humidity") + "%");
                System.out.println("Location: " + data.getDouble("latitude") + ", " + data.getDouble("longitude"));
                System.out.println("QoS: " + message.getQos());
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {}
        });

        subscriberClient.connect(options);
        subscriberClient.subscribe(targetTopic, subQos);

        // 添加中断检测循环
        while (running && !Thread.currentThread().isInterrupted()) {
            Thread.sleep(1000); // 避免空转
        }

        if (subscriberClient != null && subscriberClient.isConnected()) {
            subscriberClient.unsubscribe(targetTopic);
            subscriberClient.disconnectForcibly(1000);
        }
    }

    // 新增安全终止方法
    public void safeStop() {
        running = false;
        this.interrupt();
    }
}
